package com.hyl.msg;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

public class KfkConsumer {
	KafkaConsumer<byte[], byte[]> _consumer;
	private String _namesrvAddr = "localhost";
	private String _groupid = "tests";

	public static KfkConsumer getInstance(String groupid, String serverurl) {
		return new KfkConsumer(groupid, serverurl);
	}

	private KfkConsumer(String groupid, String serverurl) {
		_namesrvAddr = serverurl;
		_groupid = groupid;
		Properties props = KfkConfig.getSubConf();

		if (serverurl != null)
			props.put("bootstrap.servers", _namesrvAddr);
		if (_groupid != null)
			props.put("group.id", _groupid);
		_consumer = new KafkaConsumer<byte[], byte[]>(props);
	}

	public void pullMsg(String topics) {
		_consumer.subscribe(Arrays.asList(topics.split(";")));
		while (true) {

			ConsumerRecords<byte[], byte[]> records = _consumer.poll(100);// 设置超时
			for (ConsumerRecord<byte[], byte[]> record : records) {
				System.out.printf("offset = %d, key = %s, value = %s", record.offset(), new String(record.key()), new String(record.value()));
				System.out.println();
			}
		}
		// _consumer.commitSync();

	}

	public void close() {
		_consumer.close();
	}
}
